package com.pap.rocketmq.config.customerlistener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.pap.rocketmq.config.RocketMQConfig;
import com.pap.rocketmq.entity.Mq;
import com.pap.rocketmq.service.IMqCountService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * 消费者
 */
@Component
public class PapMqTopicListenerComponent {

    private static Logger logger = LoggerFactory.getLogger(PapMqTopicListenerComponent.class);

    @Autowired
    private IMqCountService mqCountService;

    @Autowired
    private RocketMQConfig rocketMQConfig;

    @PostConstruct
    public void defaultMQPushConsumer() {

        //消费者的组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketMQConfig.getPapTopicMQTopicGroup());

        //指定NameServer地址，多个地址以 ; 隔开
        consumer.setNamesrvAddr(rocketMQConfig.getPapRocketMQNameSrvAddr());
        try {
            //订阅PushTopic下Tag为push的消息
            consumer.subscribe(rocketMQConfig.getPapTopicMQTopic(), rocketMQConfig.getPapTopicMQTopictag());
            // 一次拉取多少条数据
            consumer.setConsumeMessageBatchMaxSize(100);
            //设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
            //如果非第一次启动，那么按照上次消费的位置继续消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.registerMessageListener(new MessageListenerConcurrently() {

                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                    ConsumeConcurrentlyStatus returnStatus = ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    try {
                        for (MessageExt messageExt : list) {
                            Boolean operationBool = operationDB(messageExt);
                            if (!operationBool) {
                                returnStatus = ConsumeConcurrentlyStatus.RECONSUME_LATER;
                            }
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                        logger.error("[{}]", e.getMessage());
                        returnStatus = ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                    return returnStatus;
                }
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }



    /**
     * 将文章信息推向 DB 等持久化
     * @param message
     * @return
     */
    private Boolean operationDB(MessageExt message) {
        Boolean operationBool = false;
        try {
            // mqCountService.updateCount();
            String messageStr = new String(message.getBody(), "UTF-8");
            ObjectMapper mapper = new ObjectMapper();
            Mq mqVO = mapper.readValue(messageStr, Mq.class);
            mqCountService.save(mqVO.getId(), mqVO.getName(), message.getMsgId());

            operationBool = true;
        } catch (Exception e) {
            // logger
        }
        return operationBool;
    }
}
